package com.yanggu.bigdata.realtime.app.dwm

import cn.hutool.core.date.DateUtil
import cn.hutool.json.JSONUtil
import com.alibaba.fastjson.JSON
import com.yanggu.bigdata.realtime.bean.{OrderWide, PaymentInfo, PaymentWide}
import com.yanggu.bigdata.realtime.common.GmallConfig.KAFKA_BROKER_LIST
import com.yanggu.bigdata.realtime.utils.KafkaUtil
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

/**
 * DWM层支付宽表
 * 需要将
 */
object PaymentWideApp {

  def main(args: Array[String]): Unit = {
    //1. 设置执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //2.1 读取Kafka中dwm层订单宽表的数据
    val orderWideDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "dwm_order_wide", "PaymentWideApp")
      .map(data => JSONUtil.toBean(data, classOf[OrderWide]))
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[OrderWide]()
        .withTimestampAssigner(new SerializableTimestampAssigner[OrderWide] {
          override def extractTimestamp(element: OrderWide, recordTimestamp: Long): Long = DateUtil.parseDateTime(element.getCreate_time).getTime
        }))

    //2.2 读取Kafka中dwd层Payment数据
    val paymentInfoDataStream = KafkaUtil.getKafkaDataStream(env, KAFKA_BROKER_LIST, "dwd_payment_info", "PaymentWideApp")
      .map(data => JSONUtil.toBean(data, classOf[PaymentInfo]))
      .assignTimestampsAndWatermarks(WatermarkStrategy.forMonotonousTimestamps[PaymentInfo]()
        .withTimestampAssigner(new SerializableTimestampAssigner[PaymentInfo] {
          override def extractTimestamp(element: PaymentInfo, recordTimestamp: Long): Long = DateUtil.parseDateTime(element.getCreate_time).getTime
        }))

    //3. 双流JOIN
    val paymentWideDataStream = paymentInfoDataStream
      .keyBy(_.getOrder_id)
      .intervalJoin(orderWideDataStream.keyBy(_.getOrder_id))
      .between(Time.seconds(- 30L), Time.minutes(1L))
      .process((left: PaymentInfo, right: OrderWide, _: ProcessJoinFunction[PaymentInfo, OrderWide, PaymentWide]#Context, out: Collector[PaymentWide]) => {
        out.collect(new PaymentWide(left, right))
      })

    paymentWideDataStream.print("paymentWideDataStream>>>>>")

    paymentWideDataStream
      .map(JSONUtil.toJsonStr(_))
      .addSink(KafkaUtil.getKafkaSink(KAFKA_BROKER_LIST, "dwm_payment_wide"))

    env.execute("PaymentWideApp Job")

  }

}
